package com.darrenchan.spark.javaapi;

import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

import java.util.Arrays;

public class StreamingWordCount {

    public static void main(String[] args) throws InterruptedException {
        SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("StreamingWordCount");
        JavaStreamingContext ssc = new JavaStreamingContext(conf, new Duration(5000));

        JavaReceiverInputDStream<String> lines = ssc.socketTextStream("hadoop000", 9999);

        JavaPairDStream<String, Integer> counts = lines.flatMap(line -> Arrays.asList(line.split(" ")).iterator()).
                mapToPair(word -> new Tuple2<>(word, 1)).
                reduceByKey((a, b) -> a + b);

        counts.print();

        ssc.start();
        ssc.awaitTermination();
    }

}
